package com.fwmagic.flink.projectcase.streamjoin;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.concurrent.TimeUnit;

public class StreamDataRightSource extends RichParallelSourceFunction<Tuple3<String,String, Long>> {

    private volatile boolean running = true;

    @Override
    public void run(SourceContext<Tuple3<String, String, Long>> ctx) throws Exception {
        //准备数据
        Tuple3[] elements = new Tuple3[]{
                Tuple3.of("a", "hangzhou", 10059000L), //[50000-60000)
                Tuple3.of("b", "beijing",  10105000L), //[100000-110000)
        };

        int count = 0;
        while (running && count < elements.length) {
            //将数据发送出去
            ctx.collect(Tuple3.of((String) elements[count].f0, (String) elements[count].f1, (Long) elements[count].f2));
            count++;
            TimeUnit.SECONDS.sleep(1);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
